package demo;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * 消费者
 * 
 * @author super
 *
 */
public class PushConsumer2 {

    /**
     * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。
     * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法
     */
    public static void main(String[] args) throws InterruptedException,
            MQClientException {
    	
        /**
         * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例
         * 注意：ConsumerGroupName需要由应用来保证唯一
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Common.ConsumerGroupName);
        consumer.setNamesrvAddr(Common.ADDR);
        
        /*
		 * 不要手贱
		 * 默认情况下不需要设置instanceName，rocketmq会使用ip@pid作为instanceName(pid代表jvm名字)
		 * 如果设置instanceName，rocketmq会使用ip@instanceName作为consumer的唯一标示，此时需要注意instanceName需要不同
		 */
        //consumer.setInstanceName("Consumber");
        
		/**
		 * CONSUME_FROM_LAST_OFFSET，默认值，表示从上次停止时的地方开始消费
		 * CONSUME_FROM_FIRST_OFFSET，从队列的头部开始消费
		 * CONSUME_FROM_TIMESTAMP，从指定的时间点开始消费
		 */
        // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
        // 如果非第一次启动，那么按照上次消费的位置继续消费
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(1);


		// 集群消费 OR 广播消费, 默认集群消息；
        // 此处集群or广播是在相同ConsumerGroupName为前提下的概念；一般情况下要保证ConsumerGroupName唯一，因为有两个不同GroupName时，会向两个Group广播消息
		// consumer.setMessageModel(MessageModel.BROADCASTING);

        /**
         * 订阅指定topic下tags分别等于TagA或TagC或TagD
         */
        consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
        
        /**
         * 订阅指定topic下所有消息
         * 注意：一个consumer对象可以订阅多个topic
         */
        consumer.subscribe("TopicTest2", "*");
        

        consumer.registerMessageListener(messageListenerConcurrently());
        
        //consumer.registerMessageListener(messageListenerOrderly());
        
		

        /**
         * Consumer对象在使用之前必须要调用start初始化，初始化一次即可
         */
        consumer.start();

        System.out.println("Consumer Started.");
        System.out.println("Consumer info is : "+consumer);

    }
    
    
        
    /**
     * 无序监听者(默认)
     * 
     * @return
     */
	static MessageListenerConcurrently messageListenerConcurrently() {

		return new MessageListenerConcurrently() {

			/**
			 * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
			 */
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

				System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size());

				MessageExt msg = msgs.get(0);
				System.out.println("MsgBody: " + msg.getTopic() + "," + msg.getTags() + "," + msg.getKeys() + "," + new String(msg.getBody()));

				// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		};
	}
    
    /**
     * 有序监听者
     * 
     * @return
     */
	static MessageListenerOrderly messageListenerOrderly() {
		// 消费者按照顺序严格有序消费，用这个有序的监听者。
		// 同一队列的消息同一时刻只能一个线程消费，可保证消息在同一队列严格有序消费
		return new MessageListenerOrderly() {
			@Override
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
				System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size());

				MessageExt msg = msgs.get(0);

				System.out.println("MsgBody: " + msg.getTopic() + "," + msg.getTags() + "," + msg.getKeys() + "," + new String(msg.getBody()));

				// return ConsumeOrderlyStatus.CONSUME_SUCCESS;
				return ConsumeOrderlyStatus.SUCCESS;
			}

		};
	}
	
}

